package com.zheng.cms.rocketmq.controller;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.MessageQueueSelector;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageQueue;
import com.alibaba.rocketmq.remoting.exception.RemotingException;

import com.aliyun.openservices.shade.com.alibaba.rocketmq.remoting.common.RemotingHelper;
import com.zheng.cms.common.constant.CmsResult;
import com.zheng.cms.common.constant.CmsResultConstant;
import com.zheng.cms.rocketmq.consumer.MyConsumer;
import com.zheng.cms.rocketmq.execute.TransactionExecuterImpl;
import com.zheng.cms.rocketmq.rocketmq.*;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.ResponseBody;

import java.io.UnsupportedEncodingException;
import java.util.List;

@Controller
@RequestMapping("/manage/rocketmq")
public class UserController {
    @Autowired
    @Qualifier("rocketMQProducer")
    private RocketMQProducer producer;

    @Autowired
    @Qualifier("myProducer")
    private MyProducer myProducer;

    @Autowired
    @Qualifier("producer")
    private Producer producer2;

    @Autowired
    @Qualifier("orderProducer")
    private OrderProducer orderProducer;

    @Autowired
    @Qualifier("transcationalProducer")
    private TranscationalProducer transcationalProducer;
//    @Autowired
//    @Qualifier("tcpMsgProducer")
//    private WithdrawMsgProducer withdrawMsgProducer;

    /**
     * Caused by: com.alibaba.rocketmq.client.exception.MQClientException: No route info of this topic, TestTopic1
     * See https://github.com/alibaba/RocketMQ/issues/44 for further details.
     **/
    @ResponseBody
    @RequestMapping(value = "/test", method = RequestMethod.GET)
    public Object test() throws MQClientException, RemotingException, MQBrokerException, InterruptedException {
        for (int i = 0; i < 100; i++) {
            Message msg = new Message("TestTopic1", "TAG1", (i + "這是spring集成").getBytes());
            SendResult result = producer.getDefaultMQProducer().send(msg);
            System.out.println(result);
            System.out.println(1);
        }
        return new CmsResult(CmsResultConstant.SUCCESS, "发送成功");
    }

    /**
     * SendResult [sendStatus=SEND_OK, msgId=C0A86F0100002A9F0000000000000342, messageQueue=MessageQueue [topic=TestTopic1, brokerName=PC-20170616GKHZ, queueId=0], queueOffset=3]
     * 发送成功
     * http://cms.zhangshuzheng.cn:2222/manage/rocketmq/test2
     * rocketmq的插件上的生产者就是myProducer   "TestTopic1"是主题,
     * "這是spring集成 "是消息主题
     * 生产者查不到，因为没有group?    明明是有producerGroup的,why?
     */
    @ResponseBody
    @RequestMapping(value = "/test2", method = RequestMethod.GET)
    public Object test2() throws MQClientException, RemotingException, MQBrokerException, InterruptedException {

        Message msg = new Message("TestTopic1", "TAG1", ("這是spring集成").getBytes());
        SendResult result = myProducer.getDefaultMQProducer().send(msg);
        System.out.println(result);
        System.out.println("发送成功");
        return new CmsResult(CmsResultConstant.SUCCESS, "发送成功");
    }

//
//    @ResponseBody
//    @RequestMapping(value = "/test3", method = RequestMethod.GET)
//    public Object test3() {
//        //Message msg = new Message("TestTopic1", "TAG1", ( "這是spring集成").getBytes());
//        withdrawMsgProducer.sendMessage("test");
//        System.out.println("发送成功");
//        return new CmsResult(CmsResultConstant.SUCCESS, "发送成功");
//    }

    /**
     * SendResult [sendStatus=SEND_OK, msgId=C0A86F0100002A9F000000000000056E, messageQueue=MessageQueue [topic=TestTopic1, brokerName=PC-20170616GKHZ, queueId=0], queueOffset=6]
     * 发送成功
     * <p>
     * 监听器没消费啊
     * rocketMQ无序队列消息
     */
    @ResponseBody
    @RequestMapping(value = "/test3", method = RequestMethod.GET)
    public Object test3() throws MQClientException, RemotingException, MQBrokerException, InterruptedException {

        Message msg = new Message("TestTopic3", "TAG1", ("test3").getBytes());
        SendResult result = producer2.getProducer().send(msg);
        System.out.println(result);
        System.out.println("发送成功");
        return new CmsResult(CmsResultConstant.SUCCESS, "发送成功");
    }

    /**
     * rocketmq发送有序队列消息
     *
     *
     *注意：是把把消息发到同一个队列（queue），不是同一个topic，默认情况下一个topic包括4个queue
     */
    @ResponseBody
    @RequestMapping(value = "/test4", method = RequestMethod.GET)
    public Object test4() throws MQClientException, RemotingException, MQBrokerException, InterruptedException,
            UnsupportedEncodingException {
        String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};

        // 发送10条消息到Topic为TopicTestOrdered，tag为tags数组按顺序取值，
        // key值为“KEY”拼接上i的值，消息内容为“Hello RocketMQ”拼接上i的值,
        // 当参数i为10时并没有排序，还是乱序的，但是有一点顺序，是不是因为    private volatile int defaultTopicQueueNums = 4;的参数问题？
        //当参数i为4的时候也还是乱序的？
        for (int i = 0; i < 10; i++) {
            int OrderId = i % 10;
            Message msg = new Message("TopicTestOrdered", tags[i % tags.length], "KEY" + i,
                    ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
            msg.setDelayTimeLevel(2);// 这里设置需要延时的等级即可,2即为延迟5S
            SendResult result = orderProducer.getProducer().send(msg, new MessageQueueSelector() {
                public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                    // arg的值其实就是orderId
                    Integer id = (Integer) arg;
                    // mqs是队列集合，也就是topic所对应的所有队列
                    int index = id % mqs.size();
                    return mqs.get(index);
                }
            }, OrderId);

            System.out.println(result);
            System.out.println("发送成功");
        }


        return new CmsResult(CmsResultConstant.SUCCESS, "发送成功");
    }


    /**
     *
     * 事务类型消息
     *
     * 执行本地事务msg = Hello RocketMQ 1
     * 执行本地事务arg = null
     * SendResult [sendStatus=SEND_OK, msgId=C0A86F0100002A9F00000000000077B4, messageQueue=MessageQueue [topic=TopicTransactionTest, brokerName=PC-20170616GKHZ, queueId=0], queueOffset=0]
     * 执行本地事务msg = Hello RocketMQ 2
     * 执行本地事务arg = null
     * ======我的操作============，失败了  -进行ROLLBACK
     * SendResult [sendStatus=SEND_OK, msgId=C0A86F0100002A9F0000000000007946, messageQueue=MessageQueue [topic=TopicTransactionTest, brokerName=PC-20170616GKHZ, queueId=1], queueOffset=0]
     *
     * 发送数据到MQ，并且处理本地事物。这里模拟了一个成功一个失败。Consumer只会接收到本地事物成功的数据。第二个数据失败了，不会被消费。
     * */
    @ResponseBody
    @RequestMapping(value = "/test5", method = RequestMethod.GET)
    public Object test5() throws InterruptedException {
        TransactionExecuterImpl tranExecuter = new TransactionExecuterImpl();
        for (int i = 1; i <= 2; i++) {
            try {
                Message msg = new Message("TopicTransactionTest", "transaction" + i, "KEY" + i,
                        ("Hello RocketMQ " + i).getBytes());
                SendResult sendResult = transcationalProducer.getProducer().sendMessageInTransaction(msg, tranExecuter, null);
                System.out.println(sendResult);

                Thread.sleep(10);
            } catch (MQClientException e) {
                e.printStackTrace();
            }
        }

        for (int i = 0; i < 100000; i++) {
            Thread.sleep(1000);
        }
        return new CmsResult(CmsResultConstant.SUCCESS, "发送成功");
    }




}
